package com.linkstec.spark;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
 
import java.io.Serializable;
 
public class demo6 {
    private static String appName = "spark.sql.demo";
    private static String master = "local[*]";
 
    public static void main(String[] args) {
        SparkSession spark = SparkSession
                .builder()
                .appName(appName)
                .master(master)
                .getOrCreate();
 
        JavaRDD<Person> peopleRDD = spark.read()
                .textFile("people.txt")
                .javaRDD()
                .map(new Function<String, Person>() {
                    /**
					 * 
					 */
					private static final long serialVersionUID = 1L;

					public Person call(String s) throws Exception {
                        String[] parts = s.split(",");
                        Person person = new Person();
                        person.setName(parts[0]);
                        person.setAge(Integer.parseInt(parts[1].trim()));
                        return person;
                    }
                });
 
        Dataset<Row> peopleDF = spark.createDataFrame(peopleRDD, Person.class);
        peopleDF.show();
    }
 
    public static class Person implements Serializable {
        /**
		 * 
		 */
		private static final long serialVersionUID = 1L;
		private String name;
        private int age;
 
        public String getName() {
            return name;
        }
 
        public void setName(String name) {
            this.name = name;
        }
 
        public int getAge() {
            return age;
        }
 
        public void setAge(int age) {
            this.age = age;
        }
    }
}
